
import java.io.*;
import java.util.*;


/*
 * @author MingLu (minglu@indiana.edu), Yicheng Feng(yichfeng@indiana.edu)
 */
public class SequentialPageRank {

    public static void main(String[] args) {

        System.out.println("*********************************************");
        System.out.println("* The part 1 of assignment#1 in B534 class  *");
        System.out.println("* SequentialPageRank Function               *");
        System.out.println("*********************************************");

        if (args.length != 4) {//Parameters not equal to 4
            String error_report = "\nUsage: "
                    + "java SequentialPageRank "
                    + "[input file name][output file name][iteration count][damping factor]\n"
                    + "e.g.: "
                    + "java SequentialPageRank pagerank.input.1000.2 pagerank_output.txt 1000 0.85";
            System.out.println(error_report);
            System.exit(-1);
        }

        int iterationCount = Integer.parseInt(args[2]);//10
        double dampingFactor = Double.parseDouble(args[3]);//0.85
        String inputFilename = args[0];//"pagerank.input";//"pagerank.input.1000.2";
        String outputFilename = args[1];//"pagerank_output.txt"; //Reading 4 Parameters
        int top_n = 10;
        double converge_thresh = 1e-9;

        try {
            long startTime = System.nanoTime();
            int num_urls = 0;
            HashMap<Integer, ArrayList<Integer>> adjacency_matrix = new HashMap<Integer, ArrayList<Integer>>();
            HashMap<Integer, Double> rank_values_table = new HashMap<Integer, Double>();

            //initialize
            num_urls = input_file_data_into_adjaceny_matrix(adjacency_matrix, inputFilename);
            System.out.println(num_urls);
            init_rank_value_table(rank_values_table, num_urls);
            //interate
            for (int i = 0; i < iterationCount; i++) {
                int converged = update_ranking_value_table(rank_values_table, num_urls, adjacency_matrix, dampingFactor, converge_thresh);
                if (converged == 1) {
                    System.out.println("Converged at iteration " + (i + 1));
                    break;
                }
                //System.out.println(rank_values_table);
            }
            //output
            BufferedWriter bw = new BufferedWriter(new FileWriter(outputFilename));
            List arrayList = new ArrayList(rank_values_table.entrySet());
            Collections.sort(arrayList, new HashmapComparator.valueComparator());
            Iterator it = arrayList.iterator();
            while (top_n > 0 && it.hasNext()) {
                Map.Entry entry = (Map.Entry) it.next();
                System.out.println(entry.getKey() + ":" + entry.getValue());
                bw.write(entry.getKey() + " : " + entry.getValue() + "\n");
                top_n--;
            }
            bw.close();

            long endTime = System.nanoTime();
            long time = (endTime - startTime) / 1000000;
            System.out.println("Time cost for PageRanking: " + time + " ms");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }//main

    public static int input_file_data_into_adjaceny_matrix(HashMap<Integer, ArrayList<Integer>> adjacency_matrix,
            String inputFilename) {

        int count = 0;
        try {
            BufferedReader br = new BufferedReader(new FileReader(inputFilename));
            String line;

            while ((line = br.readLine()) != null) {
                count++;
                ArrayList<Integer> outboundLink_list = new ArrayList<Integer>();
                String[] temp = line.split("\\s+");
                for (int i = 1; i < temp.length; i++) {
                    outboundLink_list.add(Integer.valueOf(temp[i]));
                }
                adjacency_matrix.put(Integer.valueOf(temp[0]), outboundLink_list);
            }
            br.close();
            System.out.println(adjacency_matrix);

        } catch (IOException e) {
            e.printStackTrace();
        }
        return count;
    }

    public static void init_rank_value_table(HashMap<Integer, Double> rank_values_table, int num_urls) {

        double initial_rank_value_per_page = 1.0 / (double) num_urls;
        for (int i = 0; i < num_urls; i++) {
            rank_values_table.put(i, initial_rank_value_per_page);
        }
        System.out.println(rank_values_table);
    }

    public static int update_ranking_value_table(HashMap<Integer, Double> rank_values_table, int num_urls,
            HashMap<Integer, ArrayList<Integer>> adjacency_matrix, double dampingFactor, double thresh) {

        HashMap<Integer, Double> intermediate_rvt = new HashMap<Integer, Double>();
        Iterator<Integer> iterator = adjacency_matrix.keySet().iterator();
        ArrayList<Integer> outbound_urls_list;
        Integer url_name, outbound_url;
        double intermediate_rank_value = 0.0, dangling_value = 0.0, dangling_value_per_page = 0.0;
        int outdegree_of_source_url = 0;
        int converged = 1;

        for (int i = 0; i < num_urls; i++) {
            intermediate_rvt.put(i, 0.0);
        }

        while (iterator.hasNext()) {
            url_name = ((Integer) iterator.next()).intValue();
            outbound_urls_list = adjacency_matrix.get(url_name);
            outdegree_of_source_url = outbound_urls_list.size();
            if (outdegree_of_source_url != 0) {
                for (int i = 0; i < outdegree_of_source_url; i++) {
                    outbound_url = outbound_urls_list.get(i).intValue();
                    intermediate_rank_value = intermediate_rvt.get(outbound_url) + rank_values_table.get(url_name) / (double) outdegree_of_source_url;
                    intermediate_rvt.put(outbound_url, intermediate_rank_value);
                }
            } else { //dangling nodes
                dangling_value += rank_values_table.get(url_name);
            }
        }
        dangling_value_per_page = dangling_value / (double) num_urls;
        for (int i = 0; i < num_urls; i++) {
            double new_value = dampingFactor * (intermediate_rvt.get(i) + dangling_value_per_page) + (1 - dampingFactor) * (1.0 / (double) num_urls);
            if (converged == 1 && ((new_value - rank_values_table.get(i)) > thresh || (rank_values_table.get(i) - new_value) > thresh)) {
                converged = 0;
            }
            //rank_values_table.put(i, dampingFactor * (intermediate_rvt.get(i) + dangling_value_per_page) + (1 - dampingFactor) * (1.0 / (double) num_urls));
            rank_values_table.put(i, new_value);
        }
//return 0;
        return converged;
    }
}

class HashmapComparator {

    public static class valueComparator implements Comparator {

        public int compare(Object o1, Object o2) {
            Map.Entry obj1 = (Map.Entry) o1;
            Map.Entry obj2 = (Map.Entry) o2;
            return obj2.getValue().toString().compareTo(obj1.getValue().toString());
        }
    }
}
